package org.mm.proxycache;

import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoSession;
import org.mm.proxycache.ConnectionManager.ConnectionCallback;
import org.mm.proxycache.http.HttpChunk;
import org.mm.proxycache.http.HttpUrl;
import org.mm.proxycache.http.HttpVersion;
import org.mm.proxycache.http.MutableHttpRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;

@Singleton
public class ProxyClientHandler extends IoHandlerAdapter
{

    static Logger log = LoggerFactory.getLogger( ProxyClientHandler.class );

    private static final int ClientConnectionIdleTime = 300;

    @Inject
    private ConnectionManager connectionManager;

    /*
     * (non-Javadoc)
     * 
     * @see org.apache.mina.common.IoHandlerAdapter#messageReceived(org.apache.mina.common.IoSession,
     *      java.lang.Object)
     */
    @Override
    @SuppressWarnings("unchecked")
    public void messageReceived( final IoSession clientSession, Object message ) throws Exception
    {
        if ( message instanceof HttpChunk )
        {
            HttpChunk chunk = (HttpChunk) message;
            handleChunk( clientSession, chunk );
            return;
        }

        final MutableHttpRequest request = (MutableHttpRequest) message;

        HttpUrl url = request.url();
        log.info( "Request for {} method={}", url, request.method() );

        request.version( HttpVersion.Http11 );
        request.removeHeader( "Proxy-Connection" );
        request.removeHeader( "Keep-Alive" );
        request.keepAlive( true );
        request.addHeader( "Via", "1.1 ProxyCache" );

        final String host = url.host() != null ? url.host() : request.header( "Host" );
        
        final int port = (url.port() == -1) ? 80 : url.port();
        if ( port == 80 )
            request.header( "Host", host );
        else
            request.header( "Host", host + ':' + port );

        if ( url.path() == null )
            url.setPath( "/" );
        
        url.setHost( null );
        url.setSchema( null );
        url.setPort( -1 );

        clientSession.suspendRead();

        connectionManager.getConnection( host, port, clientSession, new ConnectionCallback() {

            public void done( IoSession serverSession )
            {
                clientSession.resumeRead();
                clientSession.setAttribute( ServerSessionName, serverSession );
                serverSession.write( request );

                ConcurrentLinkedQueue<HttpChunk> pendingChunks = (ConcurrentLinkedQueue<HttpChunk>) clientSession
                        .getAttribute( PendingChunks );

                if ( pendingChunks != null )
                {
                    while ( true )
                    {
                        HttpChunk chunk = pendingChunks.poll();
                        if ( chunk == null )
                            break;

                        serverSession.write( chunk );
                    }
                }
            }

            public void error()
            {
                log.warn( "Error connecting to  {}:{}", host, port );
                ConcurrentLinkedQueue<HttpChunk> pendingChunks = (ConcurrentLinkedQueue<HttpChunk>) clientSession
                        .getAttribute( PendingChunks );

                if ( pendingChunks != null )
                    pendingChunks.clear();

                clientSession.resumeRead();
                clientSession.write( HttpResponses.getNameErrorResponse() );
                clientSession.closeOnFlush();
            }

        } );
    }

    private void handleChunk( IoSession clientSession, HttpChunk chunk )
    {
        IoSession serverSession = (IoSession) clientSession.getAttribute( ServerSessionName );

        if ( serverSession != null )
            serverSession.write( chunk );
        else
            addPendingChunk( clientSession, chunk );

        if ( chunk.isLast() )
        {
            log.debug( "Suspending read operations" );
            clientSession.suspendRead();
        }
    }

    @Override
    public void sessionOpened( IoSession session ) throws Exception
    {
        log.debug( "New connection from {}", session.getRemoteAddress() );
        session.getConfig().setReaderIdleTime( ClientConnectionIdleTime );
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.apache.mina.common.IoHandlerAdapter#sessionClosed(org.apache.mina.common.IoSession)
     */
    @Override
    public void sessionClosed( IoSession session ) throws Exception
    {
        log.debug( "Client Session closed: connected={} closing={}", session.isConnected(), session
                .isClosing() );
    }

    @Override
    public void sessionIdle( IoSession session, IdleStatus status ) throws Exception
    {
        log.info( "Client session Idle" );
        session.close();
    }

    @Override
    public void exceptionCaught( IoSession clientSession, Throwable cause ) throws Exception
    {
        log.debug( "Exception on client connection: {} - {}", cause.getClass().getSimpleName(), cause.getMessage() );
        IoSession serverSession = (IoSession) clientSession.getAttribute( ServerSessionName );
        if ( serverSession != null )
            connectionManager.release( serverSession );

        clientSession.close( true );
    }

    // ////////////////////////////////////

    @SuppressWarnings("unchecked")
    public void addPendingChunk( IoSession session, HttpChunk chunk )
    {
        session.setAttributeIfAbsent( PendingChunks, new ConcurrentLinkedQueue<HttpChunk>() );
        ((ConcurrentLinkedQueue<HttpChunk>) session.getAttribute( PendingChunks )).add( chunk );
    }

    protected static final String PendingChunks = ProxyClientHandler.class.getName()
            + ".pendingChunks";

    private static final String ServerSessionName = ProxyClientHandler.class.getName()
            + ".ServerSession";
}
